package cn._51doit.flink.day09;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 将WordCount的结果写入到MySQL中
 *
 * 将输出的数据写入到MySQL中，并且实现覆盖的功能
 * 在MySQL建表是，覆盖的key设置为主键（非空、唯一、建索引）
 */
public class WordCountToMySQL {

    public static void main(String[] args) throws Exception {

        //1.创建Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.调用Source创建DataStream
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        //3.调用Transformation(s)
        SingleOutputStreamOperator<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    collector.collect(word); //将数据输出给下一个算子使用
                }
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return Tuple2.of(word, 1);
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndOne.keyBy(t -> t.f0);
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyed.sum(1);

        //将聚合后的结果写入到MySQL中
        summed.addSink(JdbcSink.sink(
                "INSERT INTO tb_wordcount (word, counts) values (?, ?) on duplicate key update counts = ?",
                new JdbcStatementBuilder<Tuple2<String, Integer>>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Tuple2<String, Integer> input) throws SQLException {
                        preparedStatement.setString(1, input.f0);
                        preparedStatement.setInt(2, input.f1);
                        preparedStatement.setInt(3, input.f1);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(20)
                        .withBatchIntervalMs(500)
                        .build(),

                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://node-1.51doit.cn:3306/doit28")
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("123456")
                        .build()
        ));

        //5.启动
        env.execute();

    }
}
